#include "favorite_service.h"

#include <map>

#include "redis_executor/redis_executor.h"
#include "redis_executor/scoped_reply.h"

#include "redis_executor/scoped_reply.h"

using namespace std;
using namespace base;
using namespace apache::thrift;

namespace jztzp {
class ReloadRedisTask : public concurrency::Runnable {
  FavoriteServiceHandler * handler_;
  bool instant_load_;
 public:
  ReloadRedisTask(FavoriteServiceHandler * handler, bool load_now) : handler_(handler), instant_load_(load_now) {
  }

  virtual void run() {
    LOG_INFO("start loading redis.");
    // if (instant_load_ || time(NULL) % (24 * 60 * 60) <= 60 * 60) {
    if (instant_load_ || time(NULL) % (2 * 60) <= 60) {
      if (handler_->ReloadRedis()) {
        LOG_INFO("loading redis SUCCESS.");
      } else {
        LOG_ERROR("loading redis ERROR.");
      }
    }
    // handler_->timer_manager_.add(shared_ptr<concurrency::Runnable>(new ReloadRedisTask(handler_, false)), 1000 * 60 * 60);
    handler_->timer_manager_.add(shared_ptr<concurrency::Runnable>(new ReloadRedisTask(handler_, false)), 1000 * 60);
    LOG_INFO("restart load timer.");
  }
};


FavoriteServiceHandler::FavoriteServiceHandler() :
    db_pool_("user_account", "127.0.0.1:3306", "root", "") {
  shared_ptr<concurrency::PlatformThreadFactory> factory = shared_ptr<concurrency::PlatformThreadFactory>(new concurrency::PlatformThreadFactory());
  timer_manager_.threadFactory(factory);

  timer_manager_.start();
  timer_manager_.add(shared_ptr<concurrency::Runnable>(new ReloadRedisTask(this, true)), 1000);
}

FavoriteServiceHandler::~FavoriteServiceHandler() {
  timer_manager_.stop();
}

bool FavoriteServiceHandler::ReloadRedis() {
  mysqlpp::ScopedConnection cp(db_pool_, true);
  if (!cp) {
    LOG_ERROR("Failed to get a db connection!");
    return false;
  }

  mysqlpp::Query query(cp->query());
  query << "SELECT user_id, article_id FROM favorite";

  LOG_INFO("ReloadRedis() " << query.str());

  try{
    mysqlpp::UseQueryResult res = query.use();
    if (res) {
      // CommentItem comment;
      while(mysqlpp::Row row = res.fetch_row()) {
        // LOG_DEBUG("reloadRedis() " "reload comment " << row["id"] << " state=" << row["state"]);
        // CommentDbToThrift(row, &comment);
        // redis_adapter_.CommentAdd(comment);
        AddRedis(row["user_id"], row["article_id"]);
      }
    }
  } catch(const mysqlpp::Exception & e) {
    LOG_ERROR("ReloadRedis() error:" << e.what());
    return false;
  }
  return true;
}

bool FavoriteServiceHandler::AddRedis(int32_t member, int32_t article) {
  const int MAX_BUF_LEN = 64;
  char cmd_buf[MAX_BUF_LEN];
  cmd_buf[MAX_BUF_LEN - 1] = '\0';

  {
    snprintf(cmd_buf, MAX_BUF_LEN - 1, "SADD user:favorites:%d %d", member, article);
    ScopedReply reply(redis_exec_.Execute("", cmd_buf));
    if (!reply) {
      LOG_ERROR("redis add fav error, member=" << member);
      return false;
    }
  }

  {
    snprintf(cmd_buf, MAX_BUF_LEN - 1, "SADD article:favorites:%d %d", article, member);
    ScopedReply reply(redis_exec_.Execute("", cmd_buf));
    if (!reply) {
      LOG_ERROR("redis add fav error, article=" << article);
      return false;
    }
  }
  LOG_DEBUG("AddRedis() member=" << member << " article=" << article);
  return true;
}

int32_t FavoriteServiceHandler::Add(const int32_t member, const int32_t article) {
  const int MAX_BUF_LEN = 64;
  char cmd_buf[MAX_BUF_LEN];
  cmd_buf[MAX_BUF_LEN - 1] = '\0';

  {
    snprintf(cmd_buf, MAX_BUF_LEN - 1, "SADD user:favorites:%d %d", member, article);
    ScopedReply reply(redis_exec_.Execute("", cmd_buf));
    if (!reply) {
      LOG_ERROR("redis add fav error, member=" << member);
      return -1;
    }
  }

  {
    snprintf(cmd_buf, MAX_BUF_LEN - 1, "SADD article:favorites:%d %d", article, member);
    ScopedReply reply(redis_exec_.Execute("", cmd_buf));
    if (!reply) {
      LOG_ERROR("redis add fav error, article=" << article);
      return -1;
    }
  }

  mysqlpp::ScopedConnection cp(db_pool_, true);
  if (!cp) {
    LOG_ERROR("Failed to get a connection from the pool!");
    return -2;
  }

  mysqlpp::Query query(cp->query());
  query << "INSERT INTO favorite(user_id, article_id) VALUES(" << member << ", " << article << ')';
  LOG_DEBUG("db add fav " << query.str());

  bool res = false;
  try {
    res = query.exec();
  } catch (mysqlpp::Exception & e) {
    LOG_ERROR("db add fav error, member=" << member << " article=" << article);
    return -2;
  }
  return res ? 0 : -3;
}

int32_t FavoriteServiceHandler::Remove(const int32_t member, const int32_t article) {
  const int MAX_BUF_LEN = 64;
  char cmd_buf[MAX_BUF_LEN];
  cmd_buf[MAX_BUF_LEN - 1] = '\0';

  {
    snprintf(cmd_buf, MAX_BUF_LEN - 1, "SREM user:favorites:%d %d", member, article);
    ScopedReply reply(redis_exec_.Execute("", cmd_buf));
    if (!reply) {
      LOG_ERROR("redis rem fav error, member=" << member);
      return -1;
    }
  }

  {
    snprintf(cmd_buf, MAX_BUF_LEN - 1, "SREM article:favorites:%d %d", article, member);
    ScopedReply reply(redis_exec_.Execute("", cmd_buf));
    if (!reply) {
      LOG_ERROR("redis rem fav error, article=" << article);
      return -1;
    }
  }

  mysqlpp::ScopedConnection cp(db_pool_, true);
  if (!cp) {
    LOG_ERROR("Failed to get a connection from the pool!");
    return -2;
  }

  mysqlpp::Query query(cp->query());
  query << "DELETE FROM favorite WHERE user_id=" << member <<" AND article_id=" << article;
  LOG_DEBUG("db rem fav " << query.str());

  bool res = false;
  try {
    res = query.exec();
  } catch (mysqlpp::Exception & e) {
    LOG_ERROR("db rem fav error, member=" << member << " article=" << article);
    return -2;
  }
  return res ? 0 : -3;
}

void FavoriteServiceHandler::GetMemberFavoriteArticles(std::vector<int32_t> & _return, const int32_t member) {
  const int MAX_BUF_LEN = 64;
  char cmd_buf[MAX_BUF_LEN];
  cmd_buf[MAX_BUF_LEN - 1] = '\0';
  {
    snprintf(cmd_buf, MAX_BUF_LEN - 1, "SMEMBERS user:favorites:%d", member);
    ScopedReply reply(redis_exec_.Execute("", cmd_buf));
    if (!reply) {
      LOG_ERROR("GetMemberFavoriteArticles error: "<< cmd_buf);
      return;
    }
    if (reply->type == REDIS_REPLY_ARRAY) {
      LOG_DEBUG(cmd_buf << ", reply array size :" << reply->elements);
      for(size_t i = 0; i < reply->elements; ++i) {
        redisReply * el = reply->element[i];
        if (el->type == REDIS_REPLY_STRING) {
          int v = 0;
          sscanf(el->str, "%d", &v);
          LOG_DEBUG("id=" << v);
          _return.push_back(v);
        }
      }
    }
  }
}

void FavoriteServiceHandler::GetArticleFavoriteMembers(std::vector<int32_t> & _return, const int32_t article) {
  const int MAX_BUF_LEN = 64;
  char cmd_buf[MAX_BUF_LEN];
  cmd_buf[MAX_BUF_LEN - 1] = '\0';
  std::cout << "----------" ;
  {
    snprintf(cmd_buf, MAX_BUF_LEN - 1, "SMEMBERS article:favorites:%d", article);
    ScopedReply reply(redis_exec_.Execute("", cmd_buf));
    if (!reply) {
      LOG_ERROR("GetArticleFavoriteMembers error: "<< cmd_buf);
      return;
    }

    if (reply->type == REDIS_REPLY_ARRAY) {
      LOG_DEBUG(cmd_buf << ", reply array size :" << reply->elements);
      for(size_t i = 0; i < reply->elements; ++i) {
        redisReply * el = reply->element[i];
        if (el->type == REDIS_REPLY_STRING) {
          int v = 0;
          sscanf(el->str, "%d", &v);
          LOG_DEBUG("id=" << v);
          _return.push_back(v);
        }
      }
    }
  }
}


/*
void ArticleCommentHandler::Post(const CommentItem& item) {
  mysqlpp::ScopedConnection cp(db_pool_, true);
  if (!cp) {
    LOG_ERROR("Failed to get a connection from the pool!");
    return;
  }

  mysqlpp::Query query(cp->query());
  int64_t id = GenNextCommentId();
  if (id < 0) {
    LOG_ERROR("GenNextCommentId() error id : " << id);
    return;
  }
  const_cast<CommentItem&>(item).__set_id(id);

  redis_adapter_.CommentAdd(item);
  query << "INSERT INTO article_comment(id, article_id, sender_id, sender_name, sender_ip, sent_at, content) VALUES (" 
        << id << ',' << item.article_id << ',' << item.sender_id << ',' 
        << mysqlpp::quote << item.sender_name << ',' << mysqlpp::quote_only << item.sender_ip << ','
        << "FROM_UNIXTIME("<< item.time_stamp << ")," << mysqlpp::quote << item.content << ")";

  bool res = false;
  try {
    res = query.exec();
  } catch (mysqlpp::Exception & e) {
    LOG_ERROR("post comment " << item.id << " error:" << e.what());
  }
  LOG_DEBUG("post comment " << item.id << " success=" << res);
}

void ArticleCommentHandler::SetAuditState(const int64_t id, const AuditState::type audit) {
  mysqlpp::ScopedConnection cp(db_pool_, true);
  if (!cp) {
    LOG_ERROR("Failed to get a connection from the pool!");
    return;
  }
  mysqlpp::Query query(cp->query());
  query << "UPDATE article_comment SET state = state&(~" << g_ArticleComment_constants.kAuditStateMask << ")|" << audit << " WHERE " << "id=" << id;

  LOG_DEBUG("SetAuditState() " << query.str());

  CommentItem comment;
  if (0 == GetCommentById(&comment, id)) {
    if (audit == AuditState::AUDITED_FAIL) {
      redis_adapter_.CommentRemove(id, comment.article_id, comment.sender_id);
    } else {
      // TODO: 最好重新加载一次进来
      redis_adapter_.CommentAdd(comment);
    }
  }

  bool res = false;
  try {
    res = query.exec();
  } catch (mysqlpp::Exception & e) {
    LOG_ERROR(query.str() << " error:" << e.what());
  }
  LOG_DEBUG("SetAuditState() " << query.str() << " success=" << res);
}

void ArticleCommentHandler::Delete(const int64_t id) {
  CommentItem comment;
  if (GetCommentById(&comment, id) < 0) {
    return;
  }
  redis_adapter_.CommentRemove(id, comment.article_id, comment.sender_id);

  mysqlpp::ScopedConnection cp(db_pool_, true);
  if (!cp) {
    LOG_ERROR("Failed to get a connection from the pool!");
    return;
  }

  mysqlpp::Query query(cp->query());
  query << "DELETE FROM article_comment WHERE " << "id=" << id;

  bool res = false;
  try {
    res = query.exec();
  } catch (mysqlpp::Exception & e) {
    LOG_ERROR(query.str() << " error:" << e.what());
  }
  LOG_DEBUG("Delete() " << query.str() << " success=" << res);
}

void ArticleCommentHandler::GetById(CommentItem& _return, const int64_t id) {
  mysqlpp::ScopedConnection cp(db_pool_, true);
  if (!cp) {
    LOG_ERROR("Failed to get a connection from the pool!");
    return;
  }

  mysqlpp::Query query(cp->query());
  query << "SELECT id, state, article_id, sender_id, sender_name, UNIX_TIMESTAMP(sent_at) AS ts, content FROM article_comment WHERE " 
        << "id=" << id;
  mysqlpp::StoreQueryResult res;
  try {
    res = query.store();
    if (res) {
      if (res.num_rows() > 0) {
        CommentDbToThrift(res[0], &_return);
      } else {
        _return.id = 0;
        LOG_DEBUG("GetById() " << "id " << id << " not exist.");
      }
    } else {
      _return.id = -1;
      LOG_DEBUG("GetById() null res.");
    }
  } catch (mysqlpp::Exception & e) {
    LOG_ERROR(query.str() << " error:" << e.what());
  }
  LOG_DEBUG("GetById() " << query.str() << " success=" << res);
}

bool ArticleCommentHandler::ReloadRedis() {
  mysqlpp::ScopedConnection cp(db_pool_, true);
  if (!cp) {
    LOG_ERROR("Failed to get a connection from the pool!");
    return false;
  }

  mysqlpp::Query query(cp->query());
  query << "SELECT id, state, state, article_id, sender_id, sender_name, sender_ip, UNIX_TIMESTAMP(sent_at) AS ts, content FROM article_comment"
        << " WHERE state&0x03!=" << AuditState::AUDITED_FAIL;
  LOG_INFO("ReloadRedis() " << query.str());

  try{
    mysqlpp::UseQueryResult res = query.use();
    if (res) {
      CommentItem comment;
      while(mysqlpp::Row row = res.fetch_row()) {
        LOG_DEBUG("reloadRedis() " "reload comment " << row["id"] << " state=" << row["state"]);
        CommentDbToThrift(row, &comment);
        redis_adapter_.CommentAdd(comment);
      }
    }
  } catch(const mysqlpp::Exception & e) {
    LOG_ERROR("ReloadRedis() error:" << e.what());
    return false;
  }
  return true;
}


void ArticleCommentHandler::GetComments(CommentListResult& _return, 
    // const AuditState::type audit,
    const int32_t audit_flag,
    const int32_t offset, 
    const int32_t limit) {
  mysqlpp::ScopedConnection cp(db_pool_, true);
  if (!cp) {
    LOG_ERROR("Failed to get a connection from the pool!");
    return;
  }

  mysqlpp::Query query(cp->query());
  query << "SELECT id, state, state, article_id, sender_id, sender_name, sender_ip, UNIX_TIMESTAMP(sent_at) AS ts, content FROM article_comment";
  if (audit_flag < AuditState::AUDIT_STATE_MAX) {
    query << " WHERE state&0x03=" << audit_flag;
  }
  query << " ORDER BY id DESC LIMIT " << (offset + limit);
  LOG_DEBUG(query.str());

  try{
    mysqlpp::UseQueryResult res = query.use();
    if (res) {
      int pos = 0;
      CommentList comments;
      while(mysqlpp::Row row = res.fetch_row()) {
        if (++pos > offset) {
          comments.push_back(CommentItem());
          CommentDbToThrift(row, &comments.back());
        }
      }
      _return.__set_comments(comments);
    }
    LOG_INFO("GetComments() comment-count=" << _return.comments.size());
    _return.__set_success(true);
  } catch(const mysqlpp::Exception & e) {
    _return.__set_success(false);
    _return.__set_reason(e.what());
    LOG_ERROR("GetComments() error=" << e.what());
  }
}

void ArticleCommentHandler::GetUserComments(CommentListResult& _return, 
    const int64_t sender_id, 
    const int32_t offset, 
    const int32_t limit) {
  mysqlpp::ScopedConnection cp(db_pool_, true);
  if (!cp) {
    LOG_ERROR("Failed to get a connection from the pool!");
    return;
  }

  mysqlpp::Query query(cp->query());
  query << "SELECT id, state, article_id, sender_id, sender_name, sender_ip, UNIX_TIMESTAMP(sent_at) AS ts, content FROM article_comment"
        << " WHERE sender_id=" << sender_id << " ORDER BY id DESC LIMIT " << (offset + limit);

  LOG_DEBUG(query.str());

  try{
    mysqlpp::UseQueryResult res = query.use();
    if (res) {
      int pos = 0;
      CommentList comments;
      while(mysqlpp::Row row = res.fetch_row()) {
        if (++pos > offset) {
          comments.push_back(CommentItem());
          // CommentItem & item = comments.back();
          CommentDbToThrift(row, &comments.back());
        }
      }
      _return.__set_comments(comments);
    }
    LOG_INFO("comment-list-size=" << _return.comments.size());
    _return.__set_success(true);
  } catch(const mysqlpp::Exception & e) {
    _return.__set_success(false);
    _return.__set_reason(e.what());
    LOG_ERROR("error:" << e.what());
  }
} 

int32_t ArticleCommentHandler::GetArticleCommentCount(const int64_t article_id) {
  int32_t count = redis_adapter_.CommentCountByArticle(article_id);
  if (count >= 0) {
    return count;
  }
  mysqlpp::ScopedConnection cp(db_pool_, true);
  if (!cp) {
    LOG_ERROR("Failed to get a connection from the pool!");
    return -1;
  }

  mysqlpp::Query query(cp->query());
  query << "SELECT COUNT(id) FROM article_comment WHERE article_id="
        << article_id << " AND state!=" << AuditState::AUDITED_FAIL;
  LOG_DEBUG("GetArticleCommentCount " << article_id << query.str());
  mysqlpp::StoreQueryResult res;
  try {
    res = query.store();
  
    if (res) {
      if (res.num_rows() > 0) {
        LOG_INFO("GetArticleCommentCount " << article_id << " comment-count=" << res[0][0]);
        return res[0][0];
      } else {
        LOG_INFO("GetArticleCommentCount " << article_id << " comment-count=zero");
        return 0;
      }
    } else {
      LOG_WARN("GetArticleCommentCount " << article_id << " null res.");
      return -1;
    }
  } catch (mysqlpp::Exception & e) {
    LOG_ERROR("GetArticleCommentCount " << article_id << " error:" << e.what());
  }
  return -1;
}

void ArticleCommentHandler::GetArticleComments(CommentListResult& _return, 
    const int64_t article_id, 
    const int32_t offset, 
    const int32_t limit) {
  CommentList comments;
  if (redis_adapter_.CommentGetByArticle(&comments, article_id, offset, limit) == 0) {
    _return.__set_comments(comments);
    _return.__set_success(true);
    return;
  }

  comments.clear();

  mysqlpp::ScopedConnection cp(db_pool_, true);
  if (!cp) {
    LOG_ERROR("Failed to get a connection from the pool!");
    return;
  }

  mysqlpp::Query query(cp->query());
  query << "SELECT id, state, article_id, sender_id, sender_name, sender_ip, UNIX_TIMESTAMP(sent_at) AS ts, content FROM article_comment WHERE " 
        << "article_id=" << article_id << " AND state!=" << AuditState::AUDITED_FAIL << " ORDER BY id ASC LIMIT " << (offset + limit);

  try {
    mysqlpp::UseQueryResult res = query.use();
    if (res) {
      int pos = 0;
      while(mysqlpp::Row row = res.fetch_row()) {
        if (++pos > offset) {
          comments.push_back(CommentItem());
          // CommentItem & item = comments.back();
          CommentDbToThrift(row, &comments.back());
        }
      }
      _return.__set_comments(comments);
    }
    LOG_INFO("GetArticleComments " << article_id << " comment count = " << _return.comments.size());
    _return.__set_success(true);
  } catch(const mysqlpp::Exception & e) {
    _return.__set_success(false);
    _return.__set_reason(e.what());
    LOG_ERROR("GetArticleComments " << article_id << " error:" << e.what());
  } catch (const std::exception & e) {
    _return.__set_success(false);
    _return.__set_reason(e.what());
    LOG_ERROR("GetArticleComments " << article_id << " std::exception:" << e.what());
  }
}

int ArticleCommentHandler::GetCommentById(CommentItem * comment, int32_t id) {
  mysqlpp::ScopedConnection cp(db_pool_, true);
  if (!cp) {
    LOG_ERROR("Failed to get a connection from the pool!");
    return -1;
  }

  mysqlpp::Query query(cp->query());
  query << "SELECT id, state, article_id, sender_id, sender_name, sender_ip, UNIX_TIMESTAMP(sent_at) AS ts, content FROM article_comment WHERE " 
        << "id=" << id;

  mysqlpp::StoreQueryResult res;
  try {
    res = query.store();
  
    if (res) {
      if (res.num_rows() > 0) {
        CommentDbToThrift(res[0], comment);
        return 0;
      } else {
        LOG_INFO("Comment " << id << " not exist");
        return -1;
      }
    } else {
      LOG_ERROR("get comment " << id << " null res.");
      return -1;
    }
  } catch (mysqlpp::Exception & e) {
    LOG_ERROR("Get Comment " << id << " ex=" << e.what());
  }
  return -1;
}

void ArticleCommentHandler::CommentDbToThrift(const mysqlpp::Row & row, CommentItem * item) {
  item->__set_id(row["id"]);
  item->__set_state(row["state"]);
  item->__set_article_id(row["article_id"]);
  item->__set_sender_id(row["sender_id"]);
  if (row["sender_name"].c_str() != NULL) {
    item->__set_sender_name(row["sender_name"].c_str());
  } else {
    item->__set_sender_name("访客");
  }
  item->__set_time_stamp(row["ts"]);
  item->__set_sender_ip(row["sender_ip"].c_str());
  if (row["content"].c_str() != NULL) {
    item->__set_content(row["content"].c_str());
  } else {
    item->__set_content("");
  }
}

int64_t ArticleCommentHandler::GenNextCommentId() {
  if (base_commnet_id_ < 0) {
    mysqlpp::ScopedConnection cp(db_pool_, true);
    if (!cp) {
      LOG_ERROR("Failed to get a connection from the pool!");
      return -1;
    }
    mysqlpp::Query query(cp->query());
    query << "SELECT MAX(id) FROM article_comment";
    mysqlpp::StoreQueryResult res;
    try {
      res = query.store();
    
      if (res) {
        if (res.num_rows() > 0) {
          base_commnet_id_ = res[0][0];
          LOG_INFO(query.str() << " base_commnet_id_=" << res[0][0]);
        } else {
          base_commnet_id_ = 0;
        }
      } else {
        LOG_ERROR(query.str() << " null res.");
        return -1;
      }
    } catch (mysqlpp::Exception & e) {
      LOG_ERROR(query.str() << " error:" << e.what());
      return -1;
    }
  }

  return base_commnet_id_ + (++comment_id_offset_);
}
*/

}

int main(int argc, char **argv) {
  LOG_INIT("./favorite.log", "DEBUG");
  int port = 9095;

  LOG_INFO("listens on port " << port);
  using namespace jztzp;
  ThriftService<FavoriteServiceHandler, FavoriteServiceProcessor> service;
  service.Start(port);
  return 0;
}

/*
extern "C" {

void ServiceEntry() {
  int port = 9092;

  using namespace jztzp;
  ThriftService<ArticleCommentHandler, ArticleCommentProcessor> service;
  service.Start(port);
}

}
*/


